package cn.com.sparkle.paxos;

import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.Set;

import cn.com.sparkle.paxos.addprocess.AddRequestDealer;
import cn.com.sparkle.paxos.addprocess.speedcontrol.SpeedControlModel;
import cn.com.sparkle.paxos.deamon.CatchUpDeamon;
import cn.com.sparkle.paxos.deamon.CheckMasterSenatorStateDeamon;
import cn.com.sparkle.paxos.deamon.HeatBeatCheckDeamon;
import cn.com.sparkle.paxos.deamon.InstanceExecutor;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.net.ClientNetHandler;
import cn.com.sparkle.paxos.net.ServerNetHandler;
import cn.com.sparkle.paxos.net.factory.NetClient;
import cn.com.sparkle.paxos.net.factory.NetFactory;
import cn.com.sparkle.paxos.net.factory.NetServer;
import cn.com.sparkle.paxos.userserver.UserHandlerInterface;
import cn.com.sparkle.paxos.userserver.UserServerHandler;

public class ClassicPaxosServer {
	private Configuration conf;
	private EventsManager eventsManager = new EventsManager();
	private ClusterState cState;
	private ReConnectThread reConnectThread = new ReConnectThread();
	private Long lastExceptInstanceId = null;
	public ClassicPaxosServer(Long lastExceptInstanceId){
		this.lastExceptInstanceId = lastExceptInstanceId;
	}
	public ClassicPaxosServer(){
		this(null);
	}
	
	public void init(String filePath, UserHandlerInterface userHandlerInterface)
			throws Throwable {
		this.conf = new Configuration(filePath, eventsManager);
		cState = new ClusterState(eventsManager, conf);
		ConfigNode configNode = getSelfConfig();
		ServerNetHandler handler = tryBind(configNode);
		InstanceExecutor ie = new InstanceExecutor(userHandlerInterface, conf,
				eventsManager);
		// initiate account book
		AccountBook aBook = new AccountBook(conf,lastExceptInstanceId);
		// initiate instance executor thread
		ie.start(aBook);
		aBook.initInstanceExecutor(ie);
		// initialize server handler
		handler.setaBook(aBook);
		handler.setcState(cState);
		// handler.setAddRequestDealer(new AddRequestDealer(cState, conf, aBook,
		// eventsManager));
		// initialize speed control
		SpeedControlModel speedControlModel = new SpeedControlModel(
				eventsManager, conf);
		// connect all node
		ClientNetHandler phandler = connectNodes();
		// initiate heart beat deamon
		HeatBeatCheckDeamon beatCheckDeamon = new HeatBeatCheckDeamon(cState,
				eventsManager);
		beatCheckDeamon.start();
		// initiate master check deamon
		Thread checkSenatorDeamon = new Thread(
				new CheckMasterSenatorStateDeamon(cState, conf, eventsManager));
		checkSenatorDeamon.setName("checkSenatorDeamon");
		checkSenatorDeamon.start();
		// initiate catch up deamon
		Thread catchUpDeamon = new Thread(new CatchUpDeamon(eventsManager,
				cState, conf, aBook));
		catchUpDeamon.setName("catchUpDeamon");
		catchUpDeamon.start();
		reConnectThread.startThread();
		// initiate user server
		userHandlerInterface.setAddRequestDealer(new AddRequestDealer(cState,
				conf, aBook, eventsManager));
		tryBindUserServer(userHandlerInterface,configNode);
	}

	public EventsManager getEventsManager() {
		return eventsManager;
	}

	private ClientNetHandler connectNodes() throws Throwable {
		NetClient client = NetFactory.makeClient(conf.getNetLayer());
		ClientNetHandler phandler = new ClientNetHandler(client, eventsManager,
				reConnectThread);
		client.init(conf.getFilePath() + "/system_out_net.prop", phandler);
		

		NodesCollection senators = cState.getSenators();
		NodesCollection followers = cState.getFollowers();
		for (Entry<String, NodeState> s : senators.getNodeStates().entrySet()) {
			String[] temp = s.getKey().split(":");
			String ip = temp[0];
			String port = temp[1];
			client.connect(ip, Integer.parseInt(port), s.getValue());
		}
		for (Entry<String, NodeState> s : followers.getNodeStates().entrySet()) {
			String[] temp = s.getKey().split(":");
			String ip = temp[0];
			String port = temp[1];
			client.connect(ip, Integer.parseInt(port), s.getValue());
		}
		return phandler;
	}

	private ConfigNode getSelfConfig() throws SocketException{
		HashSet<String> allIps = new HashSet<String>();
		Enumeration<NetworkInterface> interfaceEnum = NetworkInterface
				.getNetworkInterfaces();
		while (interfaceEnum.hasMoreElements()) {
			NetworkInterface i = interfaceEnum.nextElement();
			Enumeration<InetAddress> ipEnum = i.getInetAddresses();
			while (ipEnum.hasMoreElements()) {
				allIps.add(ipEnum.nextElement().getHostAddress());
			}
		}
		Set<ConfigNode> set = conf.getNodes();
		
		for (ConfigNode s : set) {
			if (allIps.contains(s.getIp())
					&& String.valueOf(conf.getPort()).equals(s.getPort())) {
				conf.setSelfAddress(s.getIp() + ":" + s.getPort());
				return s;
			}
		}
		
		throw new RuntimeException(
				"fail to find self address,check configuration's file please.");
	}
	private ServerNetHandler tryBind(ConfigNode configNode) throws Throwable {
		ServerNetHandler handler = new ServerNetHandler(eventsManager, conf);
		NetServer server = NetFactory.makeServer(conf.getNetLayer());
		server.init(conf.getFilePath() + "/system_in_net.prop", handler);
		server.listen(configNode.getIp(), Integer.parseInt(configNode.getPort()));
		return handler;
	}

	private UserServerHandler tryBindUserServer(
			UserHandlerInterface userHandlerInterface,ConfigNode configNode) throws Throwable {
		UserServerHandler handler = new UserServerHandler(eventsManager, conf,
				userHandlerInterface);
		NetServer server = NetFactory.makeServer(conf.getNetLayer());
		server.init(conf.getFilePath() + "/service_in_net.prop", handler);
		server.listen(configNode.getIp(), Integer.parseInt(configNode.getClientPort()));
		return handler;
	}
}
